package com.winning.pmph.schedule;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.winning.pmph.entity.*;
import com.winning.pmph.enumeration.KnowledgeSystem;
import com.winning.pmph.mapper.PlatformKnowledgeLibMapper;
import com.winning.pmph.service.*;
import com.winning.pmph.utils.ApplicationContextUtil;
import com.winning.pmph.utils.LocalDateTimeUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.File;
import java.net.URL;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@Component
@ConditionalOnProperty(prefix = "scheduling", name = "enabled")
public class SyncSchedule {

    @Resource
    private KnowledgeLibService knowledgeLibService;
    @Resource
    private WordsService wordsService;
    @Resource
    private TagService tagService;
    @Resource
    private PlatformService platformService;
    @Resource
    private PlatformKnowledgeLibMapper platformKnowledgeLibMapper;
    @Resource
    private RocketMQProducerService rocketMQProducerService;
    @Resource
    private ActivityItemService activityItemService;
    @Resource
    private KnowledgeService knowledgeService;
    @Resource
    private ActivityService activityService;

    /**
     * 定时消费死信队列, 数据量很小;
     */
    @Scheduled(cron = "0 0 1 * * ?") // 每天凌晨一点启动定时任务
    public void dlqSchedule() {
        log.info("启动定时任务消费死信队列");
        List<Platform> platforms = platformService.list();
        for (Platform platform : platforms) {
            String dlq = "%DLQ%" + platform.getId() + "_CG";
            log.info("消费死信队列: " + dlq);
			DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(dlq + "_Consumer");
			consumer.setNamesrvAddr(ApplicationContextUtil.getProperty("mq.addr"));
			//一次最大消费的条数
			consumer.setPullBatchSize(100);
			//无消息时，最大阻塞时间。默认1分钟 单位ms
			consumer.setPollTimeoutMillis(60 * 1000);
			try {
				consumer.subscribe(dlq, "*");
				//要消费的topic，可使用tag进行简单过滤
				consumer.start();
				//拉取消息，无消息时会阻塞
				while (true) {
					List<MessageExt> msgs = consumer.poll();
					if (CollectionUtils.isEmpty(msgs)) {
						log.info("死信队列(" + dlq + ")无新消息, 消费组正常退出");
						consumer.shutdown();
						break;
					}
					//业务处理
					msgs.forEach(msg -> {
						String topic = platform.getId();
						String msgBody = new String(msg.getBody());
						log.info("消费死信队列:" + msgBody);
						KnowledgeSyncStatus knowledgeSyncStatus = new KnowledgeSyncStatus();
						String knowledgeId = "";
						// 知识和医学计算器 tag 均为knowledge
						if (msg.getTags().contains(KnowledgeSystem.KNOWLEDGE.value)) {
							// 知识
							JSONObject jsonObject = JSONObject.parseObject(msgBody);
							JSONObject knowledge = (JSONObject) jsonObject.get("data");
							knowledgeId = knowledge.getString("id");
							knowledgeSyncStatus.setKnowledgeLibId(knowledge.getString("knowledgeLibId"));
							knowledgeSyncStatus.setVersion(knowledge.getInteger("version"));
						} else if (msg.getTags().contains(KnowledgeSystem.JOURNAL.value)) {
							// 期刊
							JSONObject jsonObject = JSONObject.parseObject(msgBody);
							// 如果journal_id为空，则说明是期刊信息，否则是期刊文章信息
							knowledgeId = StringUtils.isBlank(jsonObject.getString("journal_id")) ? jsonObject.getString("id") : jsonObject.getString("journal_id");
						} else if (msg.getTags().contains(KnowledgeSystem.BOOK.value)) {
							// book_metadata,book_article为jsonobject ,book_catalog 为jsonarray
							if (msg.getTags().equals(KnowledgeSystem.BOOK_CATALOG.value)) {
								JSONArray jsonArray = JSON.parseArray(msgBody);
								knowledgeId = jsonArray.getJSONObject(0).getString("book_id");
							} else {
								// 如果book_id为空，则是图书文章信息，否则是图书信息
								JSONObject jsonObject = JSONObject.parseObject(msgBody);
								knowledgeId = StringUtils.isBlank(jsonObject.getString("book_id")) ? jsonObject.getString("id") : jsonObject.getString("book_id");
							}
						}
						knowledgeSyncStatus.setKnowledgeId(knowledgeId);
						knowledgeSyncStatus.setPlatformId(topic);
                        knowledgeSyncStatus.setKind(KnowledgeSyncStatus.FAILED_KIND_CONSUMER);
                        knowledgeSyncStatus.setOperatedTime(LocalDateTimeUtil.getNowAsString());
                        knowledgeLibService.saveKnowledgeSyncStatus(knowledgeSyncStatus);
					});
					consumer.commitSync();
				}
			}catch (Exception e){
				consumer.shutdown();
				log.error("消费死信队列失败!", e);
			}
		}
    }

	@Scheduled(cron = "0 0 1 * * ?") // 每天凌晨一点启动定时任务
	public void tagSchedule(){
		log.info("启动定时任务, 同步昨天修改的tag");
		log.info("tag定时任务时间: " + LocalDateTime.now().toString());
		List<Tag> updatedTags = tagService.getBaseMapper().selectBeansByWherePart("updated_time > '" + LocalDate.now().plusDays(-1) + "' order by level, `index`");
		Map<String, List<Tag>> tagListMap = updatedTags.stream().collect(Collectors.groupingBy(Tag::getRootId));
		for (String topTagId : tagListMap.keySet()) {
			// 查看当前顶级标签关联的平台;
			List<PlatformKnowledgeLib> platformKnowledgeLibs = platformKnowledgeLibMapper.executeMultiSelect(
					"SELECT distinct pkl.PLATFORM_ID " +
							" FROM platform_knowledge_lib pkl " +
							" JOIN knowledge_lib_field klf ON pkl.KNOWLEDGE_LIB_ID = klf.KNOWLEDGE_LIB_ID " +
							" WHERE klf.KIND = " + KnowledgeLibField.FIELD_KIND_REFERENCE + " AND klf.KIND_VALUE = '" + topTagId + "'");
			// 遍历平台, 同步集合数据;
			for (PlatformKnowledgeLib pkl : platformKnowledgeLibs) {
                tagListMap.get(topTagId).forEach(item -> rocketMQProducerService.sendTagToTopic(pkl.getPlatformId(), item));
			}
		}
	}

    @Scheduled(cron = "0 0 1 * * ?") // 每天凌晨一点启动定时任务
    public void knowledgeTermTagSchedule() {
		wordsService.knowledgeAndTerm();
	}

    public void getMessage() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("contribute_Consumer");
        consumer.setNamesrvAddr(ApplicationContextUtil.getProperty("mq.addr"));
        //一次最大消费的条数
        consumer.setConsumeMessageBatchMaxSize(1);
        //无消息时，最大阻塞时间。默认1分钟 单位ms
        consumer.setConsumeTimeout(10 * 1000);
        consumer.subscribe("contribute", "*");

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            ActivityItem activityItem = null;
            try {
                //业务处理
//                msgs.forEach(msg -> {
                String msgBody = new String(msgs.get(0).getBody());
                //String msgBody = "{\"knowledgeLibId\":\"0001AA100000000WD6YX\",\"wordUrl\":\"http://192.168.107.210:8080/upload/book/catalog/5595827169708032/5595827700877312/origin/5595840470599681.docx\",\"bookCatalogName\":\"慢性胃炎\",\"knowledgeLibName\":\"人卫健康知识库\",\"bookName\":\"同步测试\",\"bookCatalogId\":\"5595827700877312\",\"bookId\":\"5595827169708032\"}";

                JSONObject jsonObject = JSONObject.parseObject(msgBody);
                String knowledgeLibId = jsonObject.getString("knowledgeLibId");
                String filePath = jsonObject.getString("wordUrl");
                String fileName = jsonObject.getString("bookCatalogName");
                String knowledgeLibName = jsonObject.getString("knowledgeLibName");
                String bookName = jsonObject.getString("bookName");
                String bookCatalogId = jsonObject.getString("bookCatalogId");
                String bookId = jsonObject.getString("bookId");
                //查询此投稿活动中有
                activityItem = activityItemService.lambdaQuery().eq(ActivityItem::getId, bookCatalogId).one();
                Activity activity = new Activity();
                activity.setId(bookId);
                activity.setKnowledgeLibId(knowledgeLibId);
                activity.setName(bookName);
                activityService.saveOrUpdate(activity);
                if (activityItem == null) {
                    activityItem = new ActivityItem();
                    activityItem.setId(bookCatalogId);
                    activityItem.setActivityId(bookId);
                    activityItem.setKnowledgeName(fileName);
                    activityItem.setManuscriptName(fileName + "word");
                    activityItem.setKnowledgeLibId(knowledgeLibId);
                    activityItem.setPath(filePath);
                    activityItem.setDeleted(ActivityItem.DELETED_NO);
                    activityItemService.saveOrUpdate(activityItem);
}
                String substring = filePath.substring(filePath.lastIndexOf("/") + 1);
                String path = ApplicationContextUtil.BASE_DOWN_FORDER + "file/temp/" + substring;
                File file = new File(path);
                FileUtils.copyURLToFile(new URL(filePath), file);
                knowledgeService.importTempKnowledge(path, fileName, knowledgeLibId, activityItem);

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            } catch (Exception e) {
                log.error("消费失败!", e);
                e.printStackTrace();
                activityItem.setStatus(ActivityItem.FAIL);
                activityItem.setFailureReason(e.toString());
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            } finally {
                activityItemService.saveOrUpdate(activityItem);
            }
        });

        consumer.start();
    }
}
